package com.bwie.service.impl;

import com.baomidou.mybatisplus.extension.plugins.pagination.Page;

import com.bwie.config.KafkaConfig;
import com.bwie.mapper.TbInstitutionMapper;
import com.bwie.mapper.TbNetworkMapper;
import com.bwie.mapper.TbReportMapper;
import com.bwie.pojo.TbInstitution;
import com.bwie.pojo.TbNetwork;
import com.bwie.pojo.TbReport;
import com.bwie.service.TbNetworkService;
import com.bwie.utils.R;
import com.bwie.vo.PageInfoVo;
import org.springframework.data.redis.core.StringRedisTemplate;

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Date;
import java.util.concurrent.TimeUnit;

@Service
public class TbNetworkServiceImpl implements TbNetworkService {
    @Resource
    TbNetworkMapper tbNetworkMapper;
    @Resource
    TbInstitutionMapper tbInstitutionMapper;
    @Resource
    TbReportMapper tbReportMapper;
    @Resource
    StringRedisTemplate stringRedisTemplate;

    @Resource
    KafkaTemplate kafkaTemplate;

    @Override
    public R getNetworkList(PageInfoVo pageInfoVo) {
        Page<TbNetwork> page = new Page<>(pageInfoVo.getPageNum(), pageInfoVo.getPageSize());
        Page<TbNetwork> tbNetworkPage = tbNetworkMapper.selectPage(page, null);
        return R.OK(tbNetworkPage);
    }

    @Override
    public R getInstitutionList(PageInfoVo pageInfoVo) {
        Page<TbInstitution> page = new Page<>(pageInfoVo.getPageNum(), pageInfoVo.getPageSize());
        Page<TbInstitution> tbInstitutionPage = tbInstitutionMapper.selectInstitutionPage(page, null);


        return R.OK(tbInstitutionPage);
    }

    @Override
    public R getReportList(PageInfoVo pageInfoVo) {
        Page<TbReport> page = new Page<>(pageInfoVo.getPageNum(), pageInfoVo.getPageSize());
       Page<TbReport> tbReportPage = tbReportMapper.selectReportList(page,null);
        return R.OK(tbReportPage);
    }

    @Override
    public R addReport(TbReport tbReport) {
        tbReport.setCreateTime(new Date());
        tbReportMapper.insert(tbReport);
        String reportKey = "上报列表数据--";
        stringRedisTemplate.opsForValue().set(reportKey,tbReport.toString(),5, TimeUnit.MINUTES);
        String message = "添加上报数据成功，刷新Redis";
        System.out.println(message);
        kafkaTemplate.send(KafkaConfig.LOG_TOPIC,message);
        return R.OK();
    }


    @KafkaListener(topics = KafkaConfig.LOG_TOPIC )
    public void FollowUser(String message, Acknowledgment acknowledgment){
        //接收消息
        try {
            System.out.println("接受到消息"+message);

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            acknowledgment.acknowledge();
        }
    }

}
